package com.xjd.transport.impl;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import com.xjd.transport.Channel;
import com.xjd.transport.TransportException;
import com.xjd.transport.TransportRuntimeException;
import com.xjd.transport.TransportServer;
import com.xjd.transport.TransportStates;
import com.xjd.transport.impl.ChannelImpl.TransportAdapter;

public class TransportServerImpl implements TransportServer, TransportStates {

	protected Object stateLock = new Object();
	protected int state = STATE_INITIAL;
	protected int finishState = FINISH_STATE_SUCCESS;
	protected Throwable cause;
	protected boolean stopping = false;
	protected boolean doStopped = false;

	protected InetSocketAddress bindAddress;
	protected ExecutorService bossExecutorService;
	protected ExecutorService workerExecutorService;
	protected int maxConnectedChannel;
	protected int maxWaitingChannel;

	protected ReentrantReadWriteLock channelLock = new ReentrantReadWriteLock();
	protected Collection<ChannelImpl> connectedChannels = new HashSet<ChannelImpl>();
	protected BlockingQueue<ChannelImpl> waitingChannels = new LinkedBlockingQueue<ChannelImpl>();

	protected void checkStarted() {
		if (state >= STATE_STARTED) {
			throw new TransportRuntimeException("server already started.");
		}
	}

	protected void checkNotStart() {
		if (state <= STATE_INITIAL) {
			throw new TransportRuntimeException("server does not start.");
		}
	}

	protected void checkStopped() {
		if (state >= STATE_STOPED) {
			throw new TransportRuntimeException("server already stopped.");
		}
	}

	@Override
	public boolean isRunning() {
		synchronized (stateLock) {
			return state > STATE_INITIAL && state < STATE_STOPED;
		}
	}

	@Override
	public Throwable getCause() {
		return cause;
	}

	@Override
	public void setBindPort(int port) {
		this.setBindAddress(new InetSocketAddress(port));
	}

	@Override
	public void setBindAddress(String host, int port) {
		this.setBindAddress(new InetSocketAddress(host, port));
	}

	@Override
	public void setBindAddress(InetSocketAddress bindAddress) {
		checkStarted();
		this.bindAddress = bindAddress;
	}

	@Override
	public InetSocketAddress getBindAddress() {
		return this.bindAddress;
	}

	@Override
	public void setBossExecutorService(ExecutorService bossExecutorService) {
		checkStarted();
		this.bossExecutorService = bossExecutorService;
	}

	@Override
	public ExecutorService getBossExecutorService() {
		return this.bossExecutorService;
	}

	@Override
	public void setWorkerExecutorService(ExecutorService workerExecutorService) {
		checkStarted();
		this.workerExecutorService = workerExecutorService;
	}

	@Override
	public ExecutorService getWorkerExecutorService() {
		return this.workerExecutorService;
	}

	@Override
	public void setMaxConnectedChannel(int maxConnected) {
		checkStarted();
		this.maxConnectedChannel = maxConnected;
	}

	@Override
	public int getMaxConnectedChannel() {
		return this.maxConnectedChannel;
	}

	@Override
	public void setMaxWaitingChannel(int maxWaiting) {
		checkStarted();
		this.maxWaitingChannel = maxWaiting;
	}

	@Override
	public int getMaxWaitingChannel() {
		return this.maxWaitingChannel;
	}

	@Override
	public Channel nextWaitChannel(long timeoutInMills) {
		Channel c = null;
		if (timeoutInMills <= 0) {
			try {
				c = this.waitingChannels.take(); // 可被interrupt
			} catch (InterruptedException e) {
				// DoNothing
			}
		} else {
			long start = System.currentTimeMillis();
			long remain = timeoutInMills;
			while (remain > 0) {
				try {
					c = this.waitingChannels.poll(remain, TimeUnit.MILLISECONDS);
					break;
				} catch (InterruptedException e) {
					// DoNothing
				}
				remain = timeoutInMills - (System.currentTimeMillis() - start);
			}
		}
		synchronized (channelLock) {
			channelLock.notifyAll();
		}
		return c;
	}

	@Override
	public Channel[] getConnectedChannels() {
		synchronized (channelLock) {
			return this.connectedChannels.toArray(new Channel[0]);
		}
	}

	@Override
	public Channel[] getWaitingChannels() {
		synchronized (channelLock) {
			return this.waitingChannels.toArray(new Channel[0]);
		}
	}

	@Override
	public Channel[] getConnectedChannelsFrom(InetAddress from) {
		List<Channel> cs = new ArrayList<Channel>();
		synchronized (channelLock) {
			for (Channel c : this.connectedChannels) {
				if (c.getRemoteAddress().getAddress().equals(from)) {
					cs.add(c);
				}
			}
		}
		return cs.toArray(new Channel[0]);
	}

	@Override
	public void start() throws TransportException {
		checkStarted();
		try {
			selector = Selector.open();
			server = ServerSocketChannel.open();
			server.socket().bind(getBindAddress());
			server.configureBlocking(false);
			server.register(selector, SelectionKey.OP_ACCEPT);
			selectorThread = new SelectorThread();
			state = STATE_STARTED;
			selectorThread.start();
		} catch (Exception e) {
			throw new TransportException(e);
		}
	}

	@Override
	public void stop() {
		synchronized (stateLock) {
			if (state >= STATE_STOPED) {
				return;
			}
			checkNotStart();
			checkStopped();
			state = STATE_STOPED;
			if (cause != null) {
				finishState = FINISH_STATE_FAILED;
			} else {
				finishState = FINISH_STATE_SUCCESS;
			}
			stopping = true;
			selector.wakeup();

		}
		try {
			selectorThread.join();
		} catch (InterruptedException e) {
			// DoNothing
		}
		doStop();
	}

	protected void doStop() {
		synchronized (stateLock) {
			if (doStopped) {
				return;
			}
			doStopped = true;
			try {
				selector.close();
			} catch (IOException e) {
				// DoNothing
			}
			bossExecutorService.shutdownNow();
			try {
				server.close();
			} catch (IOException e) {
				// DoNothing
			}
			while (true) {
				ChannelImpl client = waitingChannels.poll();
				if (client == null) {
					break;
				}
				client.close();
			}
		}
	}

	Selector selector;
	ServerSocketChannel server;
	SelectorThread selectorThread;

	public class SelectorThread extends Thread {

		public void run() {

			try {
				while (!stopping) {
					int n = selector.select();
					if (n > 0) {
						for (Iterator<SelectionKey> it = selector.selectedKeys().iterator(); it.hasNext();) {
							SelectionKey key = it.next();
							it.remove();

							if (key.isValid() && key.isAcceptable()) {
								key.interestOps(key.interestOps() ^ SelectionKey.OP_ACCEPT);
								bossExecutorService.execute(new AcceptWork(server, key));
							}
						}
					}
				}
			} catch (IOException e) {
				cause = e;
			}
			doStop();
		}

	}

	public class AcceptWork extends Thread {
		ServerSocketChannel server;
		SelectionKey key;

		public AcceptWork(ServerSocketChannel server, SelectionKey key) {
			super();
			this.server = server;
			this.key = key;
		}

		public void run() {

			try {
				synchronized (channelLock) {
					while (connectedChannels.size() >= getMaxConnectedChannel() || waitingChannels.size() >= getMaxWaitingChannel()) {
						try {
							channelLock.wait();
						} catch (InterruptedException e) {
							key.interestOps(key.interestOps() & SelectionKey.OP_ACCEPT);
							return;
						}
					}
				}
				SocketChannel client = server.accept();
				ChannelImpl channel = new ChannelImpl(client, workerExecutorService, new ServerTransportAdapter());
				synchronized (channelLock) {
					connectedChannels.add(channel);
					waitingChannels.offer(channel);
				}
				key.interestOps(key.interestOps() & SelectionKey.OP_ACCEPT);
			} catch (IOException e) {
				cause = e;
				stopping = true;
			}

		}
	}

	public class ServerTransportAdapter implements TransportAdapter {

		@Override
		public void channelClosed(ChannelImpl channel) {
			synchronized (channelLock) {
				connectedChannels.remove(channel);
				waitingChannels.remove(channel);
			}
			synchronized (stateLock) {
				if (state >= STATE_STOPED) {
					synchronized (channelLock) {
						if (connectedChannels.size() == 0) {
							workerExecutorService.shutdown();
						}
					}
				}
			}
		}

	}

}
